#include <errno.h>
#include <sys/socket.h>
#include <netinet/tcp.h>

#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h>

#include <poll.h>

#include <arpa/inet.h>

#include "SocketConnPool.h"


LoggerPtr SocketConnPool::_logger = Logger::getLogger("SocketConnPool");

int SocketConnPool::init(const char *ip, const unsigned short port, const int conn_timeout, const int pool_size) {
	if (NULL == ip || port < 1 || pool_size < 1) {
		LOG4CXX_ERROR(_logger, "socket conn pool init error, invalid args");
		return -1;
	}
	_ip = ip;
	_port = port;
	_conn_timeout = conn_timeout;
	_pool_size = pool_size;
	for (int i = 0; i < pool_size; i++) {
		SOCKET_CONN_POOL_ITEM item;
		item.fd = -1;
		item.is_in_use = false;
		item.is_network_error = false;
		_conns.push_back(item);
	}
	pthread_mutex_init(&_mutex, NULL);
	LOG4CXX_INFO(_logger, "socket conn pool init ok, ip:" << ip << ", port:" << port << ", conn_timeout:" << conn_timeout << \
			", pool_size:" << pool_size);
	return 1;
}


int SocketConnPool::get(SocketConnPtr & ptr) {
	pthread_mutex_lock(&_mutex);
	int i = 0;
	for (vector<SOCKET_CONN_POOL_ITEM>::iterator iter = _conns.begin(); _conns.end() != iter; iter++, i++) {
		if (iter->is_in_use) continue;
		if (iter->fd > 0 && !iter->is_network_error) {
			int serr(0);
                	socklen_t len = sizeof(serr);
                	int ret = getsockopt(iter->fd, SOL_SOCKET, SO_ERROR, &serr, &len);
			if (0 == ret && 0 == serr) {
				char c;
				int r = recv(iter->fd, &c, 1, MSG_PEEK);
				if (0 != r) {
					ptr.assign(iter->fd, i, this);
					iter->is_in_use = true;
					pthread_mutex_unlock(&_mutex);
					return 1;
				}
				LOG4CXX_ERROR(_logger, "found socket has been closed by perror, need to reconnect");
			}
			else {
				LOG4CXX_ERROR(_logger, "found socket error, in get socket conn in socket conn pool, fd:" << \
							iter->fd << ", err code:" << serr << ", getsockopt ret:" << ret << \
							", errno:" << errno << ", need to reconnect");
			}
                }
		if (iter->fd < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", reconnect due to socket is not connected, fd:" << iter->fd);
		}
		else if (iter->is_network_error) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", reconnect due to socket encountered network error previously, "
					"fd:" << iter->fd);
		}
		else {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", reconnect due to socket has been detected in error now, fd:" << \
					iter->fd);
		}	
		int ret = reconnect(i);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, "reconnect error, ip:" << _ip << ", port:" << _port << ", ret:" << ret);
			pthread_mutex_unlock(&_mutex);
			return -1;
		}
		ptr.assign(iter->fd, i, this);
		iter->is_in_use = true;
		iter->is_network_error = false;
		pthread_mutex_unlock(&_mutex);
		return 2;
	}
	LOG4CXX_ERROR(_logger, "all the socket connection in pool are in use, could not get socket connection, ip:" << _ip << \
				", port:" << _port << ", pool_max_size:" << _pool_size);
	pthread_mutex_unlock(&_mutex);
	return -2;
}

int SocketConnPool::reconnect(int index) {
	if (index < 0) {
		LOG4CXX_ERROR(_logger, "reconnect error, invalid index:" << index);
		return -1;
	}
	int fd = socket(PF_INET, SOCK_STREAM, 0);
	if (fd < 0) {
		LOG4CXX_ERROR(_logger, "in reconnect, could not create new socket");
		return -2;
	}
	int flags = fcntl(fd, F_GETFL, 0);
	int ret = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
	if (0 != ret) {
		LOG4CXX_ERROR(_logger, "reconnect error, fail to set socket to non-block, ret:" << ret);
		close(fd);
		return -3;
	}
	struct sockaddr_in sin;
        memset(&sin, 0, sizeof(sin));
        sin.sin_family = AF_INET;
	sin.sin_port = htons(_port);
        ret = inet_aton(_ip.c_str(), &(sin.sin_addr));
	if (0 == ret) {
		LOG4CXX_ERROR(_logger, "reconnect error, invalid ip address:" << _ip);
		close(fd);
		return -4;
	}
	ret = connect(fd, (struct sockaddr *)&sin, sizeof(sin));
	if (0 == ret) {
			int ofd = _conns[index].fd;
			if (ofd >= 0) close(ofd);
			_conns[index].fd = fd;
			_conns[index].is_in_use = false;
			_conns[index].is_network_error = false;
			LOG4CXX_INFO(_logger, "reconnect ok, ip:" << _ip << ", port:" << _port << ", index:" << index);
			return 1;
	}
	if (EINPROGRESS != errno && EINTR != errno) {
		LOG4CXX_ERROR(_logger, "connect to ip:" << _ip << " error, fd:" << fd << ", errno:" << errno);
		close(fd);
		return -5;
	}

	int no_delay_flag = 1;
	setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &no_delay_flag,  sizeof(no_delay_flag));

	struct pollfd p;
	bzero(&p, sizeof(p));
	p.fd = fd;
	p.events = POLLOUT;
	int t = -1;
	if (_conn_timeout > 0) t = 1000 * _conn_timeout;
	ret = poll(&p, 1, t);
	if (0 == ret){
		LOG4CXX_ERROR(_logger, "connection timeout on socket:" << fd << ", ip:" << _ip << ", port:" << _port << \
				", conn_timeout:" << _conn_timeout);
		close(fd);
		return -6;
	}
	if (-1 == ret) {
		LOG4CXX_ERROR(_logger, "poll error in reconnect, fd:" << fd << ", ip:" << _ip << ", port:" << _port << \
				", errno:" << errno);
		return -7;
	}
	if (POLLOUT != p.revents) {
		LOG4CXX_ERROR(_logger, "in reconnect, an error has happend on socket:" << fd << ", revents:" << p.revents << \
				", ip:" << _ip << ", port:" << _port);
		close(fd);
		return -8;
	}
	int serr(0);
	socklen_t len = sizeof(serr);
	getsockopt(fd, SOL_SOCKET, SO_ERROR, &serr, &len);
	if (0 != serr) {
		LOG4CXX_ERROR(_logger, "found socket error, in new connection, fd:" << fd << ", err code:" << serr << \
				", ip:" << _ip << ", port:" << _port);
		close(fd);
		return -9;
	}
	int ofd = _conns[index].fd;
	if (ofd >= 0) close(ofd);
	_conns[index].fd = fd;
	_conns[index].is_in_use = false;
	_conns[index].is_network_error = false;
	LOG4CXX_INFO(_logger, "reconnect to ip:" << _ip << ", port:" << _port << ", index:" << index << ", ok");
	return 2;
}

int SocketConnPool::release(int index, bool is_network_error) {
	if (index < 0) {
		LOG4CXX_ERROR(_logger, "release error, invalid index:" << index);
		return -1;
	}
	pthread_mutex_lock(&_mutex);
	_conns[index].is_in_use = false;
	_conns[index].is_network_error = is_network_error;
	pthread_mutex_unlock(&_mutex);
	if (is_network_error) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", has been told network error, index:" << index); 
	}
	return 1;
}
